package cn.com.guage.flink.transformation;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 
* @version:1.0.1
* @Description: reduce算子
* reduce算子是flink流处理中的一个聚合算子，可以对属于同一个分组的数据进行一些聚合操作。
* @author: yangdechao
* @date: datedate 2021年11月12日 下午3:14:54
 */
public class ReduceTransformation {
	public static void main(String[] args) throws Exception {
		
		
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        
        DataStreamSource<String> source = environment.fromElements("a", "b", "c", "d","a");
       
        SingleOutputStreamOperator<Tuple3<String, Integer,String>> map = source.map(new MapFunction<String, Tuple3<String, Integer,String>>() {
       
        	
        	/**
			 * 
			 */
			private static final long serialVersionUID = -8228904553889423260L;

			public Tuple3<String, Integer,String> map(String value) throws Exception {
                Tuple3<String, Integer,String> tuple2 = new Tuple3<String, Integer, String>();
                tuple2.f0 = value;
                tuple2.f1 = 1;
                tuple2.f2 = "z";
                return tuple2;
            }
        });
 
        /**
         * reduce在按照同一个Key分组的数据流上生效，它接受两个输入，生成一个输出，
         * 即两两合一地进行汇总操作，生成一个同类型的新元素。
         */
        map.keyBy(0).reduce(new RichReduceFunction<Tuple3<String, Integer,String>>() {
            /**
			 * 
			 */
			private static final long serialVersionUID = -3684634381080763788L;

					@Override
                    public Tuple3<String, Integer,String> reduce(Tuple3<String, Integer,String> value1, Tuple3<String, Integer,String> value2) throws Exception {
                        String s = "f";
                        value1.f1=value1.f1+value2.f1;
                        value1.f2=s;
                        return value1;
                    }
                }).print();
 
        environment.execute();
 
    }
}
